package com.leyantech.ecp.agenx.common.redis;

import com.leyantech.ecp.agenx.common.di.RedissonModule;
import com.leyantech.ecp.agenx.common.misc.CircuitBreakerFactory;
import com.leyantech.ecp.agenx.common.redis.entity.OnlineAssistantSession;
import com.leyantech.utility.config.ConfigUtil;
import com.leyantech.utility.constructs.Constructs;
import com.leyantech.utility.statsd.StatsDClientFactory;

import com.google.common.base.Preconditions;
import com.timgroup.statsd.StatsDClient;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBucket;
import org.redisson.api.RMap;
import org.redisson.api.RRateLimiter;
import org.redisson.api.RSet;
import org.redisson.api.RedissonClient;

import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-29.
 */
@Slf4j
public class RedissonOperation<R> {

  protected final String operationKey;

  protected final Function<String, R> function;

  private final Supplier<CircuitBreaker> circuitBreaker;

  private static final RedissonClient redissonClient = RedissonModule.redissonClient;

  /**
   * 店铺在线客服，rpc 接口从该 map 查询，客服登录登出、配置变更需要维护该 map.
   */
  public static final RedissonOperation<RMap<String, OnlineAssistantSession>> STORE_ASSISTANT = RedissonOperation
      .map("StoreAssistant");

  /**
   * 客服所在节点，用来处理长链接服务停机、重启，在线客服数据刷新.
   */
  public static final RedissonOperation<RSet<String>> NODE_ASSISTANT = RedissonOperation
      .set("NodeAssistant");

  /**
   * 商品推荐卡片，客户端会上报，误触二次催下单，过滤用.
   */
  public static final RedissonOperation<RBucket<String>> ITEM_CARD = RedissonOperation
      .bucket("ItemCard");

  /**
   * connector 停机、重启 锁，避免重复发送 kafka(重复发也没事，但没必要).
   */
  public static final RedissonOperation<RBucket<String>> SHUTDOWN_LOCK = RedissonOperation
      .bucket("ShutdownLock");

  static {
    ITEM_CARD.executor("warmup").run(RBucket::isExists, "isExists");
  }

  private RedissonOperation(String operationKey, Function<String, R> function) {
    this.operationKey = operationKey;
    this.function = function;
    this.circuitBreaker = CircuitBreakerFactory.circuitBreakerConfig(operationKey);
  }

  public static <O> RedissonOperation<RMap<String, O>> map(String operationKey) {
    return new RedissonOperation<>("Map." + operationKey, redissonClient::getMap);
  }

  public static RedissonOperation<RRateLimiter> rateLimiter(String operationKey) {
    return new RedissonOperation<>("RateLimiter." + operationKey,
        redissonClient::getRateLimiter);
  }

  public static <O> RedissonOperation<RBucket<O>> bucket(String operationKey) {
    return new RedissonOperation<>("Bucket." + operationKey, redissonClient::getBucket);
  }

  public static <O> RedissonOperation<RSet<O>> set(String operationKey) {
    return new RedissonOperation<>("Set." + operationKey, redissonClient::getSet);
  }

  protected R get(String key) {
    return function.apply(RedisKeyUtils.keyOf(this.operationKey, key));
  }

  public RedisExecutor<R> executor(String key) {
    return new RedisExecutor<>(key, this);
  }

  public static class RedisExecutor<R> {

    private final String key;

    private final R r;

    private final RedissonOperation<R> operation;

    public RedisExecutor(String key, RedissonOperation<R> operation) {
      this.key = key;
      this.operation = operation;
      this.r = operation.get(key);
    }

    public void run(Consumer<R> consumer, String statsName) {
      execute(r -> {
            consumer.accept(r);
            return null;
          }, null, statsName
      );
    }

    public <O> O get(Function<R, O> fun, String statsName) {
      return execute(fun, null, statsName);
    }

    public <O> O get(Function<R, O> fun, Supplier<O> fallback, String statsName) {
      return execute(fun, fallback, statsName);
    }

    static final String STATS_PREFIX = String
        .join(".", "redis", Constructs.getInstanceNoOrDefault());

    static final Supplier<Boolean> enableKeyStats = ConfigUtil.getDynamicConfig(
        "redis",
        "key.stats.enable",
        Boolean::parseBoolean,
        Boolean.FALSE
    );

    <O> O execute(Function<R, O> func, Supplier<O> fallback, String statsName) {
      Preconditions.checkArgument(StringUtils.isNotBlank(statsName), "statsName is blank.");

      final StatsDClient client = StatsDClientFactory.getClient();
      final String operationKey = operation.operationKey;
      final String statsKey = enableKeyStats.get()
          ? String.join(".", STATS_PREFIX, operationKey, statsName, key)
          : String.join(".", STATS_PREFIX, operationKey, statsName);

      final CircuitBreaker circuitBreaker = operation.circuitBreaker.get();
      final long start = System.nanoTime();
      try {
        circuitBreaker.acquirePermission();

        final O res = func.apply(r);

        final long duration = System.nanoTime() - start;
        client.time(statsKey, duration);
        client.increment(statsKey + ".succeed");

        circuitBreaker.onResult(duration, TimeUnit.NANOSECONDS, res);
        log.debug("redis operation {} {} key {} res {} succeed, duration_ms {}",
            operationKey, statsName, key, res, TimeUnit.NANOSECONDS.toMillis(duration));

        return res;
      } catch (Throwable throwable) {
        final long duration = System.nanoTime() - start;
        client.increment(statsKey + ".error");

        circuitBreaker.onError(duration, TimeUnit.NANOSECONDS, throwable);

        log.error("redis operation {} key {} failed", operationKey, key, throwable);

        if (fallback != null) {
          return fallback.get();
        }
        throw throwable;
      }
    }

  }

}
